<?php
$serv = new swoole_server('0.0.0.0', 9509);
$serv->set(array(
    'daemonize' => 1,
    'worker_num' => 8, //worker进程数量
    'task_worker_num' => 8, //task进程数量 即为维持的MySQL连接的数量
    'log_file' => '/var/www/swoole/cli/log/9509.log',
));

function my_onReceive($serv, $fd, $from_id, $data)
{
    echo "收到数据".$data.PHP_EOL;
    $result = $serv->taskwait($data,100);
//    $result = $serv->taskwait($data,);
    echo "任务结束".PHP_EOL;
    if ($result !== false) {
        list($status, $db_res) = explode(':', $result, 2);
        if ($status == 'OK') {
            //数据库操作成功了，执行业务逻辑代码，这里就自动释放掉MySQL连接的占用
            //将处理结果发送给客户端
            $serv->send($fd, var_export(unserialize($db_res), true) . "\n");
        } else {
            $serv->send($fd, $db_res);
        }
        return;
    } else {
        $serv->send($fd, "Error. Task timeout\n");
    }
}

function my_onTask($serv, $task_id, $from_id, $sql)
{
    echo "开始做任务 task id:".$task_id.PHP_EOL;
    $link = @mysqli_connect("120.31.136.107", "root", "d23a048282d9f0e3", "system_pay");
    if (!$link) {
        $serv->finish("ER:" . mysqli_error($link));
        return;
    }
    try {
        $result = $link->query($sql);
    }catch(Exception $e){
        $link = mysqli_connect("120.31.136.107", "root", "d23a048282d9f0e3", "system_pay");
        $result = $link->query($sql);
    }

    if(preg_match("/^select/i", $sql)){//如果是select操作，就返回关联数组
        $data = $result->fetch_assoc();
    }else{//否则直接返回结果
        $data = $result;
    }
    $serv->finish("OK:" . serialize($data));//调用finish方法，用于在task进程中通知worker进程，投递的任务已完成
    //return "OK:".serialize($data);
}

function my_onFinish($serv, $task_id, $data)
{
    echo "任务完成";//taskwait 没有触发这个函数。。
    echo "AsyncTask Finish:Connect.PID=" . posix_getpid() . PHP_EOL;
}

$serv->on('receive', 'my_onReceive');
$serv->on('task', 'my_onTask');
$serv->on('Finish', 'my_onFinish');

$serv->start();//启动server






